package com.fxd.config;

import com.fxd.enums.RedisStreamEnum;
import com.fxd.model.Fruit;
import com.fxd.util.RedisTemplateUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.*;
import org.springframework.data.redis.stream.StreamListener;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.data.redis.stream.Subscription;

import javax.annotation.Resource;
import java.time.Duration;
import java.util.HashMap;
import java.util.Map;

@Slf4j
@Configuration
public class RedisStreamConfiguration {

    @Autowired
    private RedisTemplateUtil redisUtil;

    @Resource
    private StreamListener fruitsGroupConsumer1;

    @Resource
    private StreamListener fruitsGroupConsumer2;

    @Bean
    public Subscription fruitsGroupConsumer1Sub(RedisConnectionFactory factory) {
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, ObjectRecord<String, Fruit>>
                options = StreamMessageListenerContainer
                .StreamMessageListenerContainerOptions
                .builder()
                .pollTimeout(Duration.ofSeconds(1))
                .targetType(Fruit.class)
                .build();
        StreamMessageListenerContainer<String, ObjectRecord<String, Fruit>>
                container = StreamMessageListenerContainer.create(factory, options);
        //生成者Stream的key和消费者组需要提前创建
        initStream(RedisStreamEnum.FRUITS_PRODUCER_STREAM.getValue(), RedisStreamEnum.FRUITS_CONSUMER_GROUP.getValue());
        Subscription subscription = container.receive(Consumer.from(RedisStreamEnum.FRUITS_CONSUMER_GROUP.getValue(), RedisStreamEnum.FRUITS_GROUP_CONSUMER1.getValue()),
                StreamOffset.create(RedisStreamEnum.FRUITS_PRODUCER_STREAM.getValue(), ReadOffset.lastConsumed()), fruitsGroupConsumer1);
        container.start();  //lastConsumed，最后一个被消费的消息之后，开始消费
        return subscription;
    }

    @Bean
    public Subscription fruitsGroupConsumer2Sub(RedisConnectionFactory factory) {
        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, ObjectRecord<String, Fruit>>
                options = StreamMessageListenerContainer
                .StreamMessageListenerContainerOptions
                .builder()
                .pollTimeout(Duration.ofSeconds(1))
                .targetType(Fruit.class)
                .build();
        StreamMessageListenerContainer<String, ObjectRecord<String, Fruit>>
                container = StreamMessageListenerContainer.create(factory, options);
        //生成者Stream的key和消费者组需要提前创建
        initStream(RedisStreamEnum.FRUITS_PRODUCER_STREAM.getValue(), RedisStreamEnum.FRUITS_CONSUMER_GROUP.getValue());
        Subscription subscription = container.receive(Consumer.from(RedisStreamEnum.FRUITS_CONSUMER_GROUP.getValue(), RedisStreamEnum.FRUITS_GROUP_CONSUMER2.getValue()),
                StreamOffset.create(RedisStreamEnum.FRUITS_PRODUCER_STREAM.getValue(), ReadOffset.lastConsumed()), fruitsGroupConsumer2);
        container.start();  //lastConsumed，最后一个被消费的消息之后，开始消费
        return subscription;
    }

    private void initStream(String stream, String groupName) {
        if (!redisUtil.hashKey(stream)) {  //流不存在则创建
            Map<String, Object> map = new HashMap<>();
            map.put("initKey", "initValue");
            RecordId recordId = redisUtil.xadd(stream, map);
            //将初始化的值删除
            redisUtil.xDel(stream, recordId);
            log.info("stream: {}创建成功...", stream);
        }
        if (!redisUtil.xInfoGroups(stream).contains(groupName)) {
            redisUtil.xCreateGroup(stream, groupName);
            log.info("stream_group: {}创建成功...", groupName);
        }
    }

}
